package com.lfy.chat.service.impl;


import cn.hutool.json.JSONUtil;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.lfy.auth.domain.entity.UserDetailEntity;
import com.lfy.auth.domain.entity.UserEntity;
import com.lfy.auth.service.UserDetailService;
import com.lfy.chat.client.ChannelCache;
import com.lfy.chat.domain.bo.FriendBO;
import com.lfy.chat.domain.bo.VerifyMsgBo;
import com.lfy.chat.domain.msg.UnOlineEventMsg;
import com.lfy.chat.domain.msg.VerityEventRead;
import com.lfy.chat.domain.request.AgreeFriendReq;
import com.lfy.chat.domain.request.NoAgreeFriendReq;
import com.lfy.chat.domain.request.addFriendReq;
import com.lfy.chat.domain.request.SingleMsgReq;
import com.lfy.chat.domain.response.*;
import com.lfy.chat.domain.ws.UnOlineMsg;
import com.lfy.chat.domain.ws.req.ChatReq;
import com.lfy.chat.domain.ws.req.LoginReq;
import com.lfy.chat.domain.ws.resp.*;
import com.lfy.chat.service.BaseChatService;
import com.lfy.chat.service.ChatService;
import com.lfy.common.constant.ChatReidsConstant;
import com.lfy.common.constant.GlobalConstant;
import com.lfy.common.enums.BaseEnum;
import com.lfy.common.enums.ChatEnum;
import com.lfy.common.result.R;
import com.lfy.commons.config.thread.OrdinaryThreadFactory;
import com.lfy.commons.init.ApplicationInit;
import com.lfy.commons.utils.JsonUtils;
import com.lfy.commons.utils.JwtUtil;
import com.lfy.commons.utils.RedisUtil;
import io.jsonwebtoken.Claims;
import io.netty.channel.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import javax.annotation.Resource;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

/**
 * 聊天业务
 *
 * @author <a href="https://www.lvfeiy.top">吕飞扬</a>
 * @date 2023/10/24 14:42
 */
@Service("chatServiceImpl")
@Slf4j
public class ChatServiceImpl extends BaseChatService implements ChatService, ApplicationInit {
    @Resource
    private ApplicationEventPublisher publisher;
    @Resource
    private UserDetailService userDetailService;
    private static final ThreadPoolExecutor opThreadPool;

    private static final AtomicInteger I = new AtomicInteger(0);
    static {
        opThreadPool = OrdinaryThreadFactory.getOpThreadPool();
    }
    @Override
    public void init() {
        log.info("re-init verifyId ? >>:{} value:{}",RedisUtil.setNx(ChatReidsConstant.VERIFY_ID, GlobalConstant.STRING_1),RedisUtil.get(ChatReidsConstant.VERIFY_ID));
    }
    @Override
    public void unBind(Channel channel) {
        //TODO 群聊
//        ChannelCache.unBindGroup(id,channel);
        ChannelCache.unBinding(channel);
    }

    @Override
    public void login(Channel channel, ChatReq chatReq) {
        String data = chatReq.getData();
        LoginReq loginReq = JSONUtil.toBean(data, LoginReq.class);
        String token = loginReq.getToken();
        log.info(token);
        if (Objects.isNull(token)) {
            verifyError(channel);
            return;
        }
        Claims claims = JwtUtil.parseJwt(token);
        if (Objects.isNull(claims)) {
            verifyError(channel);
            return;

        }
        Object user = claims.get("user");
        if (Objects.isNull(user)) {
            verifyError(channel);
            return;
        }
        ObjectMapper objectMapper = new ObjectMapper();
        UserEntity userEntity = objectMapper.convertValue(user, UserEntity.class);
        ChannelCache.bind(userEntity.getId().toString(), channel);
            BaseResp<LoginResp> loginRespBaseResp = new BaseResp<>();
            loginRespBaseResp.setType(ChatEnum.VERIFY_SUCCESS.getType());
            loginRespBaseResp.setData(new LoginResp().setCode(ChatEnum.VERIFY_SUCCESS.getType()));
            sendMsg(channel,loginRespBaseResp);
    }

    @Override
    public R<Void> singleSend(SingleMsgReq msgReq, UserEntity entity) {
        String text = msgReq.getText();
        Integer to = msgReq.getTo();
        if(Objects.isNull(to)){
            return R.error();
        }
        Channel channel = ChannelCache.getChannel(to.toString());


        BaseResp<SingleSendResp> singleSendRespBaseResp = new BaseResp<>();

        singleSendRespBaseResp.setType(ChatEnum.singleSend.getType())
                .setData(new SingleSendResp()
                        .setFrom(entity.getId().toString())
                        .setText(text)
                        .setDate(formatDate(new Date())));


        if(Objects.isNull(channel)){
            //离线
            publisher.publishEvent(new UnOlineEventMsg(this,to.toString(),entity.getId().toString(),text));
            return R.success();
        }
        sendMsg(channel, singleSendRespBaseResp);


        return R.success();
    }


    @Override
    public R<List<OfflineListResp>> getOfflineMsg(UserEntity user) {
        Map<Object, Object> map = RedisUtil.hmget(ChatReidsConstant.CHAT_SINGLE_PREFIX.concat(user.getId().toString()));
        if(CollectionUtils.isEmpty(map)){
            return R.success();
        }
        List<OfflineListResp> offlineListResps = new ArrayList<>();
        map.forEach((k,v)->{
            List<UnOlineMsg> unOlineMsgs = JSONUtil.parseArray((String) v).toList(UnOlineMsg.class);
            List<OfflineResp> list = unOlineMsgs.stream().map(item -> {
                OfflineResp offlineResp = new OfflineResp();
                BeanUtils.copyProperties(item, offlineResp);
                return offlineResp;
            }).sorted(Comparator.comparing(OfflineResp::getDate)).collect(Collectors.toList());

            OfflineListResp offlineListResp = new OfflineListResp();
            offlineListResp.setId((String) k);
            offlineListResp.setOfflineRespList(list);
            offlineListResps.add(offlineListResp);
        });
        opThreadPool.execute(()->{
            if(map.toString().getBytes().length> GlobalConstant.BIG_KEY){
                map.forEach((k,v)->{
                    try {
                        Thread.sleep(100);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    RedisUtil.hdel(ChatReidsConstant.CHAT_SINGLE_PREFIX.concat(user.getId().toString()),k);
                });

            }else {
                RedisUtil.del(ChatReidsConstant.CHAT_SINGLE_PREFIX.concat(user.getId().toString()));
            }
        });
        return R.success(offlineListResps);
    }

    //TODO 改多线程
    @Override
    public R<Void> addFriend(UserEntity user, addFriendReq req) {
        Long from = user.getId();
        Integer to = req.getTo();
        if(from.equals(Long.parseLong(to.toString()))){
            return R.error(BaseEnum.commonEnum.ERROR_ADD_ME);
        }
        String text = req.getMsg();
        Channel channel = ChannelCache.getChannel(to.toString());
        UserDetailEntity fromDetail = userDetailService.getOne(new LambdaQueryWrapper<UserDetailEntity>().eq(UserDetailEntity::getUserId, from));
        UserDetailEntity toDetail = userDetailService.getOne(new LambdaQueryWrapper<UserDetailEntity>().eq(UserDetailEntity::getUserId, Long.parseLong(to.toString())));
        if(Objects.isNull(fromDetail) || Objects.isNull(toDetail)){
            log.error("online user :{}or{} no default userDetail",user.getId(),to);
            return R.error();
        }
        BaseResp<FriendResp> friendRespBaseResp = new BaseResp<>();
        FriendResp friendResp = new FriendResp().setId(from).setName(fromDetail.getName())
                .setImg(fromDetail.getImg())
                .setUserName(user.getUserName())
                .setText(text)
                .setSignature(fromDetail.getSignature());
        friendRespBaseResp.setType(ChatEnum.addFriend.getType()).setData(friendResp);
        //双方验证消息列表
        String i = RedisUtil.inc(ChatReidsConstant.VERIFY_ID);
        boolean aNull = Objects.isNull(channel);

        VerifyMsgBo meVerifyMsgBo = new VerifyMsgBo();
        meVerifyMsgBo.setId(to.toString()).setType(ChatEnum.waitFriend.getType()).setIsRead(ChatEnum.read.getType())
                .setText(text).setVeryMsgId(i)
                .setDateTime(formatDate(new Date()));

        VerifyMsgBo toVerifyMsgBo = new VerifyMsgBo();
        toVerifyMsgBo.setId(from.toString()).setType(ChatEnum.confirmedFriend.getType())
                .setText(text).setVeryMsgId(i)
                .setDateTime(formatDate(new Date()));
        if(aNull){
        toVerifyMsgBo.setIsRead(ChatEnum.noRead.getType());
        }else {
            toVerifyMsgBo.setIsRead(ChatEnum.read.getType());
        }

        RedisUtil.sSet(ChatReidsConstant.CHAT_VERIFY_PREFIX.concat(user.getId().toString()),meVerifyMsgBo);
        RedisUtil.sSet(ChatReidsConstant.CHAT_VERIFY_PREFIX.concat(to.toString()),toVerifyMsgBo);
        if(aNull){
            //离线
//            publisher.publishEvent(new UnOlineEventFriend(this,friendResp,to.toString()));
            return R.success();
        }

        sendMsg(channel,friendRespBaseResp);
        return R.success();
    }


    @Override
    public R<List<OfflineFriendResp>> getOfflineFriendMsg(UserEntity user) {
        List<String> strings = RedisUtil.lGet(ChatReidsConstant.CHAT_FRIEND_PREFIX.concat(user.getId().toString()), 0L, -1L);
       if(CollectionUtils.isEmpty(strings)){
          return R.success();
       }
        List<CompletableFuture<OfflineFriendResp>> futureList = strings.stream().map(item -> CompletableFuture.supplyAsync(() -> {
            OfflineFriendResp offlineFriendResp = new OfflineFriendResp();
            FriendResp friendResp = JsonUtils.toObj(item, FriendResp.class);
            BeanUtils.copyProperties(friendResp,offlineFriendResp);
            return offlineFriendResp;
        },opThreadPool)).collect(Collectors.toList());
        List<OfflineFriendResp> result = CompletableFuture.allOf(futureList.toArray(new CompletableFuture[0])).thenApply(e -> futureList.stream()
                .map(CompletableFuture::join)
                .filter(Objects::nonNull).collect(Collectors.toList())).join();

        return R.success(result);
    }

    @Override
    public R<Void> agreeFriend(UserEntity user, AgreeFriendReq agreeFriendReq) {
        String friendId = agreeFriendReq.getId();
        String msgId = agreeFriendReq.getVeryMsgId();
        UserDetailEntity friendDetail = userDetailService.getOne(new LambdaQueryWrapper<UserDetailEntity>().eq(UserDetailEntity::getUserId, friendId));

        if(Objects.isNull(friendDetail)){
            return R.error(BaseEnum.commonEnum.SYSTEM_ERROR);
        }
        UserDetailEntity meDetail = userDetailService.getOne(new LambdaQueryWrapper<UserDetailEntity>().eq(UserDetailEntity::getUserId, user.getId()));
        //判断是否是好友
        Set<String> set = RedisUtil.sGet(ChatReidsConstant.FRIEND_LIST_PREFIX.concat(user.getId().toString()));
        Set<String> friends = set.stream().map(item -> JsonUtils.toObj(item, String.class)).collect(Collectors.toSet());
        if(friends.contains(friendId)){
            return R.error(BaseEnum.commonEnum.ERROR_ADD_FRIEND);
        }
        if(Objects.isNull(meDetail)){
            return R.error(BaseEnum.commonEnum.SYSTEM_ERROR);
        }
        //验证消息修改
        R<Void> error = modifyMsg(user, friendId, msgId,ChatEnum.agreeFriend,ChatEnum.agreeFriend);
        if (error != null) {
            return error;
        }
//        FriendBO friendBO = new FriendBO();
//        friendBO.setSignature(friendDetail.getSignature()).setName(friendDetail.getName()).setId(friendDetail.getUserId().toString()).setImg(friendDetail.getImg());
        //添加进自己的好友列表
        RedisUtil.sSet(ChatReidsConstant.FRIEND_LIST_PREFIX.concat(user.getId().toString()),friendDetail.getUserId().toString());
        //对方添加进好友列表
//        FriendBO me = new FriendBO();
//        me.setImg(meDetail.getImg()).setSignature(meDetail.getSignature()).setName(meDetail.getName()).setId(user.getId().toString());
        RedisUtil.sSet(ChatReidsConstant.FRIEND_LIST_PREFIX.concat(friendId),user.getId().toString());
        Channel channel = ChannelCache.getChannel(agreeFriendReq.getId());
        //由于前端逻辑问题，如果直接用发消息的逻辑，会有好友未添加，通过axios异步请求还未返回新的好友列表时候获取不到好友头像等问题，所以这里用ws直接发送，在ws里接收时同步获取新的好友列表
        if(!Objects.isNull(channel)){
            BaseResp<HelloResp> helloRespBaseResp = new BaseResp<>();
            helloRespBaseResp.setType(ChatEnum.helloFriend.getType());
            HelloResp helloResp = new HelloResp();
            helloResp.setTime(formatDate(new Date()));
            helloResp.setText("我们已经成为好友啦~快来聊天吧");
            helloResp.setId(user.getId().toString());
            helloRespBaseResp.setData(helloResp);
            sendMsg(channel,helloRespBaseResp);
        }else {
            SingleMsgReq singleMsgReq = new SingleMsgReq();
            singleMsgReq.setText("我们已经成为好友啦~快来聊天吧");
            singleMsgReq.setTo(Integer.valueOf(friendId));
            this.singleSend(singleMsgReq,user);
        }




            //双方对话列表
            RedisUtil.sSet(ChatReidsConstant.CHAT_NOW_PREFIX.concat(user.getId().toString()),friendDetail.getUserId().toString());
            RedisUtil.sSet(ChatReidsConstant.CHAT_NOW_PREFIX.concat(friendId),user.getId().toString());


        return R.success();
    }

    /**
     * 验证消息修改
     * @param user
     * @param friendId
     * @param msgId
     * @return
     */
    private R<Void> modifyMsg(UserEntity user, String friendId, String msgId,ChatEnum userChatEnum,ChatEnum friendChatEnum) {
        //验证消息修改 TODO 维护一个索引
        synchronized ((GlobalConstant.VERIFY_LOCK.concat(friendId)).intern()){
            Set<String> userVeryMsgList = RedisUtil.sGet(ChatReidsConstant.CHAT_VERIFY_PREFIX.concat(user.getId().toString()));
            Set<String> friendVeryMsgList = RedisUtil.sGet(ChatReidsConstant.CHAT_VERIFY_PREFIX.concat(friendId));
            if(CollectionUtils.isEmpty(userVeryMsgList) || CollectionUtils.isEmpty(friendVeryMsgList)){
                return R.error();
            }
            String rollBack = null;
            for (String s : userVeryMsgList) {
                VerifyMsgBo verifyMsgBo = JsonUtils.toObj(s, VerifyMsgBo.class);
                if (verifyMsgBo.getVeryMsgId().equals(msgId)) {
                    if(!verifyMsgBo.getType().equals(ChatEnum.confirmedFriend.getType())){
                        return R.error(BaseEnum.commonEnum.DOUBLE_SET_UP);
                    }
                    verifyMsgBo.setType(userChatEnum.getType());
                    RedisUtil.setRemove(ChatReidsConstant.CHAT_VERIFY_PREFIX.concat(user.getId().toString()),s);
                    rollBack = JsonUtils.toStr(verifyMsgBo);
                    RedisUtil.sSet(ChatReidsConstant.CHAT_VERIFY_PREFIX.concat(user.getId().toString()),verifyMsgBo);
                    break;
                }
            }
            for (String s : friendVeryMsgList) {
                VerifyMsgBo verifyMsgBo = JsonUtils.toObj(s, VerifyMsgBo.class);
                if (verifyMsgBo.getVeryMsgId().equals(msgId)) {
                    if(!verifyMsgBo.getType().equals(ChatEnum.waitFriend.getType())){
                        //删除脏数据
                        String finalRollBack = rollBack;
                        opThreadPool.execute(()->{
                            RedisUtil.setRemove(ChatReidsConstant.CHAT_VERIFY_PREFIX.concat(user.getId().toString()), finalRollBack);
                            RedisUtil.setRemove(ChatReidsConstant.CHAT_VERIFY_PREFIX.concat(friendId),s);
                        });
                        return R.error(BaseEnum.commonEnum.DOUBLE_SET_UP);
                    }
                    verifyMsgBo.setType(friendChatEnum.getType());
                    verifyMsgBo.setIsRead(ChatEnum.read.getType());
                    RedisUtil.setRemove(ChatReidsConstant.CHAT_VERIFY_PREFIX.concat(friendId),s);
                    RedisUtil.sSet(ChatReidsConstant.CHAT_VERIFY_PREFIX.concat(friendId),verifyMsgBo);
                    break;
                }
            }
        }

        return null;
    }

    @Override
    public R<List<FriendBO>> getChatNowFriendList(UserEntity user) {
        Set<String> set = RedisUtil.sGet(ChatReidsConstant.CHAT_NOW_PREFIX.concat(user.getId().toString()));
        if(CollectionUtils.isEmpty(set)){
            return R.success();
        }
        List<Long> ids = set.stream().map(item-> Long.parseLong(JsonUtils.toObj(item,String.class))).collect(Collectors.toList());
        List<UserDetailEntity> detailEntities = this.userDetailService.list(new LambdaQueryWrapper<UserDetailEntity>().in(UserDetailEntity::getUserId, ids));
        List<FriendBO> result = ids.stream().map(item -> {
            FriendBO friendBO = new FriendBO();
            friendBO.setId(item.toString());
            for (UserDetailEntity detailEntity : detailEntities) {
                if (detailEntity.getUserId().equals(item)) {
                    friendBO.setImg(detailEntity.getImg())
                            .setSignature(detailEntity.getSignature())
                            .setName(detailEntity.getName());
                    break;
                }
            }
            return friendBO;
        }).collect(Collectors.toList());
        return R.success(result);
    }

    @Override
    public R<Void> removeNotChat(UserEntity user, String id) {
        opThreadPool.execute(()->{
            RedisUtil.setRemove(ChatReidsConstant.CHAT_NOW_PREFIX.concat(user.getId().toString()),JsonUtils.toStr(id));
        });
        return R.success();
    }


    @Override
    public R<Void> noAgreeFriend(UserEntity user, NoAgreeFriendReq noAgreeFriendReq) {
        R<Void> r = modifyMsg(user, noAgreeFriendReq.getId(), noAgreeFriendReq.getVeryMsgId(), ChatEnum.noAgreeFriend, ChatEnum.sideNoAgreeFriend);
        if(r != null){
            return r;
        }
        return R.success();
    }

    @Override
    public R<List<VerifyMsgResp>> getVerifyMsgList(UserEntity user) {
        Set<String> strings = RedisUtil.sGet(ChatReidsConstant.CHAT_VERIFY_PREFIX.concat(user.getId().toString()));
        if(CollectionUtils.isEmpty(strings)){
            return R.success();
        }
        Set<String> ids = new HashSet<>();
        List<VerifyMsgResp> result = strings.stream().map(item -> {
            VerifyMsgBo verifyMsgBo = JsonUtils.toObj(item, VerifyMsgBo.class);
            String veriyId = verifyMsgBo.getVeryMsgId();
            ids.add(veriyId);
            VerifyMsgResp verifyMsgResp = new VerifyMsgResp();
            BeanUtils.copyProperties(verifyMsgBo, verifyMsgResp);
            verifyMsgResp.setVeryMsgId(Integer.valueOf(veriyId));

            return verifyMsgResp;
        }).sorted(Comparator.comparing(VerifyMsgResp::getDateTime).reversed()).collect(Collectors.toList());
        Set<Long> boIds = result.stream().map(item -> Long.parseLong(item.getId())).collect(Collectors.toSet());
        List<UserDetailEntity> list = this.userDetailService.list(new LambdaQueryWrapper<UserDetailEntity>().in(UserDetailEntity::getUserId, boIds));
        for (VerifyMsgResp verifyMsgResp : result) {
            for (UserDetailEntity userDetailEntity : list) {
                if(userDetailEntity.getUserId().equals(Long.parseLong(verifyMsgResp.getId()))){
                    verifyMsgResp.setSignature(userDetailEntity.getSignature());
                    verifyMsgResp.setImg(userDetailEntity.getImg());
                    verifyMsgResp.setName(userDetailEntity.getName());
                }
            }
        }
        publisher.publishEvent(new VerityEventRead(this,ids, ChatReidsConstant.CHAT_VERIFY_PREFIX.concat(user.getId().toString()),user.getId().toString()));
        return R.success(result);
    }

    @Override
    public R<List<UserFriendResp>> getFriendList(UserEntity user) {
        Set<String> jsons = RedisUtil.sGet(ChatReidsConstant.FRIEND_LIST_PREFIX.concat(user.getId().toString()));
        if(CollectionUtils.isEmpty(jsons)){
            return R.success();
        }
        Set<String> friendIds = jsons.stream().map(item -> JsonUtils.toObj(item, String.class)).collect(Collectors.toSet());
        List<UserDetailEntity> friendEntity = this.userDetailService.list(new LambdaQueryWrapper<UserDetailEntity>().in(UserDetailEntity::getUserId, friendIds));
        List<UserFriendResp> result = friendEntity.stream().map(item -> {
            UserFriendResp userFriendResp = new UserFriendResp();
            BeanUtils.copyProperties(item, userFriendResp);
            userFriendResp.setId(item.getUserId());
            return userFriendResp;

        }).collect(Collectors.toList());
        return R.success(result);
    }

    @Override
    public R<Void> addFriendNowChat(UserEntity user, String id) {
         RedisUtil.sSet(ChatReidsConstant.CHAT_NOW_PREFIX.concat(user.getId().toString()),id);
        return R.success();
    }

    @Override
    public R<Void> delFriend(UserEntity user,String id) {
        String jsonId = JsonUtils.toStr(id);
        RedisUtil.setRemove(ChatReidsConstant.FRIEND_LIST_PREFIX.concat(user.getId().toString()),jsonId);
        RedisUtil.setRemove(ChatReidsConstant.CHAT_NOW_PREFIX.concat(user.getId().toString()),jsonId);
        CompletableFuture.runAsync(()->{
            String userId = JsonUtils.toStr(user.getId().toString());
            RedisUtil.setRemove(ChatReidsConstant.FRIEND_LIST_PREFIX.concat(id),userId);
            RedisUtil.setRemove(ChatReidsConstant.CHAT_NOW_PREFIX.concat(id),userId);
            Channel channel = ChannelCache.getChannel(id);
            if(!Objects.isNull(channel)){
                BaseResp<Void> voidBaseResp = new BaseResp<>();
                voidBaseResp.setType(ChatEnum.flushFriend.getType());
                sendMsg(channel,voidBaseResp);
            }
        },opThreadPool);


        return R.success();
    }


    public static void main(String[] args) {
        ArrayList<Integer> integers = new ArrayList<>();
        integers.add(10);
        integers.add(20);
        integers.add(30);
        Integer test =0;
        Integer add =0;
        for (Integer integer : integers) {
           integer++;
        }
        integers.forEach(System.out::println);
//        System.out.println("test"+test);
//        System.out.println("add"+add);
    }

    private void verifyError(Channel channel) {
        BaseResp<LoginResp> loginRespBaseResp = new BaseResp<>();
        loginRespBaseResp.setType(ChatEnum.VERIFY_ERROR.getType());
        LoginResp loginResp = new LoginResp();
        loginResp.setCode(ChatEnum.VERIFY_ERROR.getType());
        sendMsg(channel, loginRespBaseResp);
        channel.close();
    }


}
